
package com.gs.netty.client;
import com.gs.netty.client.remoting.WebSocketClient;

import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Channel 加权轮询
 */
public class RoundRobinLoadBalance extends AbstractLoadBalance<ChannelWrapper> {

    /**
     * key：服务接口名称
     * value： key = 提供者ip+ 服务接口名称 。value= 轮询权重实例对
     */
    private final ConcurrentMap<String, ConcurrentMap<String, WeightedRoundRobin>> methodWeightMap = new ConcurrentHashMap<>();


    @Override
    protected ChannelWrapper doLoadBalance(List<ChannelWrapper> invokerChannels) {

        /**
         * 案例，现在有3个服务提供者：
         *      服务提供者（权重值）：A（3）  B（1）  C（5）
         *      权重总和值 totalWeight： 3+1+5 =9
         *
         * 代码逻辑：
         *
         *   服务提供者（权重值,轮询权重初始值0）：   A（3,0）        B（1,0）       C（5,0）
         *              第1次 rpc 请求：   A（3,0+3=3）    B（1,0+1=1）   C（5,0+5=5）
         *                   本次调用：轮询权重最大是C，然后修改 C服务费提供者的 轮询权重 变为最小轮询权重值
         *                   轮询权重修改（权重总和值+轮询权重值）： -9+5 =-4
         *              第2次 rpc 请求：   A（3,3+3=6）    B（1,1+1=2）   C（5,-4+5=1）
         *                   本次调用：轮询权重最大是A，然后修改 A服务费提供者的 轮询权重 变为最小轮询权重值
         *                   轮询权重修改（权重总和值+轮询权重值）： -9+6 =-3
         *              第3次 rpc 请求：   A（3,-3+3=0）    B（1,2+1=3）   C（5,1+5=6）
         *                   本次调用：轮询权重最大是C，然后修改 C 服务费提供者的 轮询权重 变为最小轮询权重值
         *                   轮询权重修改（权重总和值+轮询权重值）： -9+6 =-3
         *              第4次 rpc 请求：   A（3,0+3=3）    B（1,3+1=4）   C（5,-3+5=2）
         *                   本次调用：轮询权重最大是B，然后修改 B服务费提供者的 轮询权重 变为最小轮询权重值
         *                   轮询权重修改（权重总和值+轮询权重值）： -9+4 =-5
         *              第5次 rpc 请求：   A（3,3+3=6）    B（1,-5+1=-4）   C（5,2+5=7）
         *                   本次调用：轮询权重最大是C，然后修改 C服务费提供者的 轮询权重 变为最小轮询权重值
         *                   轮询权重修改（权重总和值+轮询权重值）： -9+7 =-2
         *              第5次 rpc 请求：   A（3,6+3=9）    B（1,-4+1=-3）   C（5,-2+5=3）
         *                   本次调用：轮询权重最大是A，然后修改 A服务费提供者的 轮询权重 变为最小轮询权重值
         *                   轮询权重修改（权重总和值+轮询权重值）： -9+9 =0
         *              ....
         *
         *   上面就是代码执行逻辑，要进记住！！！
         */
        long now = System.currentTimeMillis();
        // 记录轮询最大的权重
        long maxCurrent = Long.MIN_VALUE;

        int totalWeight = 0;
        ChannelWrapper loadBalanceChannelWrapper = null;
        for (ChannelWrapper channelWrapper : invokerChannels) {

            // 轮询权重+当前服务权重
            long currentWeight = channelWrapper.increaseCurrent();

            channelWrapper.setLastUpdate(now);
            // 判断 当前服务提供者轮询权重值 是否大于 当前最大轮询权重值
            if (currentWeight > maxCurrent) {

                // 记录当前服务提供者轮询权重值 为最大轮询权重值
                maxCurrent = currentWeight;
                // 记录当前服务提供者为本次轮询胜出者
                loadBalanceChannelWrapper = channelWrapper;
            }
            // 权重相加 得到权重总和
            totalWeight += currentWeight;
        }
        if (loadBalanceChannelWrapper != null) {
            /**
             * 修改轮询权重值。轮询权重值 - 服务提供者权重总和值，为了使这个服务提供者轮询权重值变为最小轮询权重值
             */
            loadBalanceChannelWrapper.sel(totalWeight);

            return loadBalanceChannelWrapper;
        }

        return invokerChannels.get(0);
    }

    @Override
    protected boolean isAvailable(ChannelWrapper channelWrapper) {
        return channelWrapper.isAvailable();
    }



    protected static class WeightedRoundRobin {
        // 服务提供者权重值
        private int weight;

        // 轮询权重的值
        private final AtomicLong current = new AtomicLong(0);

        // 最后更新时间
        private long lastUpdate;

        public int getWeight() {
            return weight;
        }

        public void setWeight(int weight) {
            this.weight = weight;
            current.set(0);
        }

        public long increaseCurrent() {
            // 加权值
            return current.addAndGet(weight);
        }

        public void sel(int total) {
            current.addAndGet(-1 * total);
        }

        public long getLastUpdate() {
            return lastUpdate;
        }

        public void setLastUpdate(long lastUpdate) {
            this.lastUpdate = lastUpdate;
        }
    }


}
